package cn.jly.flink.source;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;

import java.net.URL;

/**
 * @author lanyangji
 * @create 2020-09-03 20:13
 */
public class CommonSourceApp {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        URL resource = CommonSourceApp.class.getClassLoader().getResource("person.csv");
        env.readCsvFile(resource.getPath())
                .pojoType(Person.class, "name", "age", "email")
                .map(new MapFunction<Person, Person>() {
                    @Override
                    public Person map(Person value) throws Exception {
                        value.setAge(value.age + 1);
                        return value;
                    }
                })
                .print();

        env.execute("CommonSourceApp");
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Person {
        private String name;
        private Integer age;
        private String email;
    }
}
